<?php

namespace app\command\workerman\process\queue;

//use app\common\service\CommonService;
use think\Exception;
use think\facade\Cache;
use Workerman\Timer;
use Workerman\Worker;

class Client
{
	private Worker|null $worker;

	/**
	 * @var object|null|\Redis
	 */
	private mixed $redis;

	/**
	 * 默认订阅选项
	 * @var array|int[]
	 */
	private array $defaultOption = [
		'max_retry'      => 3, // 最大重试次数
		'retry_interval' => 5, // 重试间隔s
		'job'            => 'job' // 执行方法名
	];

	private string $prefix = '';

	private bool $waitingStatus = false;

	private int $delayStatus = 0;

	/**
	 * 获取完整key
	 * @param $keyVarName
	 * @return string
	 */
	private function getKey($keyVarName): string
	{
		$keys = [
			// 存放所有监听的队列完整名称
			'waitingGroup' => 'SYSTEM:QUEUE:WAITING:GROUP',
			// 队列前缀
			'waiting'      => 'SYSTEM:QUEUE:WAITING:',
			// 延迟队列完整key
			'delay'        => 'SYSTEM:QUEUE:DELAY',
			// 失败存放队列完整key
			'fail'         => 'SYSTEM:QUEUE:FAIL:'
		];

		return $this->prefix . $keys[$keyVarName];
	}

	public function __construct($worker = null)
	{
		$this->worker = $worker;
		$this->redis  = Cache::store('redis')->handler();
		$this->prefix = config('cache.stores.redis.prefix');
	}

	/**
	 * 发布任务
	 * @param string   $queue
	 * @param array    $data
	 * @param int      $delay
	 * @param int|null $maxRetry
	 * @param int|null $retryInterval
	 * @return mixed
	 * @throws \RedisException
	 */
	public function send(
		string   $queue = '',
		array    $data = [],
		int      $delay = 0,
		null|int $maxRetry = null,
		null|int $retryInterval = null
	): mixed
	{
		$time    = time();
		//$id      = CommonService::getSnowFlackID();
        $id      = rand(); // 此处可替换为自己的生成算法，仅使用rand模拟
		$dataStr = $this->jsonParse([
			'id'             => $id,
			'queue'          => $queue,
			'time'           => $time,
			'data'           => $data,
			'delay'          => $delay,
			'retry'          => 0,
			'max_retry'      => $maxRetry ?? $this->defaultOption['max_retry'],
			'retry_interval' => $retryInterval ?? $this->defaultOption['retry_interval']
		]);

		if ($delay > 0) {
			$this->redis->zAdd(
				$this->getKey('delay'),
				$time + $delay,
				$dataStr
			);
			return $id;
		}

		$this->redis->lPush(
			$this->getKey('waiting') . $queue,
			$dataStr
		);
		return $id;
	}

	/**
	 * 任务订阅
	 * @param string|array $queue
	 * @param string|null  $class
	 * @param string|null  $job
	 * @return void
	 * @throws \RedisException
	 * @throws \Exception
	 */
	public function subscribe(string|array $queue = '', null|string $class = null, null|string $job = null): void
	{
		if (empty($queue)) {
			throw new \Exception('[ $queue ] Empty');
		}
		if (is_string($queue)) {
			if (empty($class)) {
				throw new \Exception('[ $class ] Empty');
			}
		} else {
			foreach ($queue as $name => $class) {
				$this->subscribe($name, $class);
			}
			return;
		}

		$this->redis->hSet(
			$this->getKey('waitingGroup'),
			$this->getKey('waiting') . $queue,
			$this->jsonParse([
				'class' => $class,
				'job'   => $job ?? $this->defaultOption['job'],
			])
		);
	}

	/**
	 * @throws Exception
	 */
	public function run(): void
	{
		try {
			$this->delayQueue();
			$this->works();
		} catch (\Exception $e) {
			throw new Exception('Queue Run Error');
		}
	}

	/**
	 * 队列任务
	 * @return void
	 * @throws \RedisException
	 */
	private function works(): void
	{
		if ($this->waitingStatus) return;

		$keys = $this->redis->hKeys($this->getKey('waitingGroup'));

		if (empty($keys)) return;

		Timer::add(0.000001, function () use ($keys) {

			$this->waitingStatus = true;

			$data = $this->redis->brPop($keys, 1);

			if (empty($data)) return;

			$key     = $data[0];
			$dataStr = $data[1];

			// 数据解析失败，放入失败队列
			try {
				$data = $this->jsonParse($dataStr);
			} catch (\Throwable|\Exception|\Error $e) {
				$this->fail(null, $dataStr);
				return;
			}
			if (!$dataStr) {
				$this->fail($data['queue'], $dataStr);
				return;
			}

			// 订阅中不存在放回队列
			if (!in_array($key, $keys)) {
				$this->redis->rPush($key, $dataStr);
				return;
			}

			// 任务执行
			try {
				$keyGetData = $this->redis->hGet(
					$this->getKey('waitingGroup'),
					$this->getKey('waiting') . $data['queue']
				);
				if (!is_string($keyGetData)) {
					return;
				}
				$group = $this->jsonParse($keyGetData);
			} catch (\Throwable|\Exception|\Error $e) {
				return;
			}
			try {

				app($group['class'])->{$group['job']}($data['data'], $data);

			} catch (\Throwable|\Exception|\Error $e) {
				if ($data['retry'] + 1 > $data['max_retry']) {
					$data['error_class']     = $group['class'];
					$data['error_class_job'] = $group['job'];
					$data['error_message']   = $e->getMessage();
					$data['error_trace']     = $e->getTraceAsString();
					$this->fail($data['queue'], $data);
				} else {
					$data['retry']++;
					$this->retry($data);
				}
			}

		});
	}

	/**
	 * 延迟任务
	 * @return void
	 */
	private function delayQueue(): void
	{
		if ($this->delayStatus > 0) return;

		$this->delayStatus = Timer::add(1, function () {
			$scoreArr = $this->redis->zRevRangeByScore($this->getKey('delay'), time(), '-inf', ['LIMIT', 0, 128]);
			if (!empty($scoreArr)) {
				foreach ($scoreArr as $dataStr) {
					$result = $this->redis->zRem($this->getKey('delay'), $dataStr);
					if ($result !== 1) {
						continue;
					}

                    try {
						$data = CommonService::jsonParse($dataStr);
					} catch (\Throwable|\Exception|\Error $e) {
						continue;
					}

					try {
						$this->redis->lPush($this->getKey('waiting') . $data['queue'], $dataStr);
					} catch (\Throwable|\Exception|\Error $e) {
						$this->fail($data['queue'], $dataStr);
						continue;
					}
				}
			}
		});
	}

	/**
	 * 放入失败队列
	 * @param string|null  $queue
	 * @param array|string $data
	 * @return void
	 * @throws \RedisException
	 */
	private function fail(string|null $queue = null, array|string $data = []): void
	{
		$this->redis->lPush(
			$this->getKey('fail') . (is_null($queue) ? 'default' : $queue),
			is_string($data) ? $data : CommonService::jsonParse($data)
		);
	}

	/**
	 * 按间隔时间进行重试
	 * @param array|string $data
	 * @return void
	 * @throws \RedisException
	 */
	private function retry(array|string $data): void
	{
		$this->redis->zAdd(
			$this->getKey('delay'),
			time() + $data['delay'] * $data['retry'],
			is_string($data) ? $data : $this->jsonParse($data)
		);
	}

	private function jsonParse($data)
	{
		return is_string($data) ? json_decode($data, JSON_UNESCAPED_UNICODE) : json_encode($data, true);
	}
}